package com.example.processfunction;

import com.example.beans.Event;
import com.example.source.ClickSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 *  
 *  * @projectName flink-1.13.2-study
 *  * @title     ProcessingTimeTimerTest   
 *  * @package    com.example.processfunction  
 *  * @description    使用处理时间定时器  
 *  * @author hjx     
 *  * @date   2022-4-25 11:15  
 *  * @version V1.0.0
 *  * @copyright 2022 ty
 *  
 */
public class ProcessingTimeTimerTest {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 处理时间语义，不需要分配时间戳和 watermark
        DataStreamSource<Event> streamSource = env.addSource(new ClickSource());

        // 要用定时器，必须基于 KeyedStream
        streamSource.keyBy(data -> true)
                .process(new MyKeyedProcessFunction())
                .print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


    private static class MyKeyedProcessFunction extends KeyedProcessFunction<Boolean, Event, String> {

        @Override
        public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
            Long currTs = ctx.timerService().currentProcessingTime();
            out.collect("数据到达，到达时间：" + new Timestamp(currTs));
            // 注册一个 10 秒后的定时器
            ctx.timerService().registerProcessingTimeTimer(currTs + 10 * 1000L);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            out.collect("定时器触发，触发时间：" + new Timestamp(timestamp));
        }
    }
}
